package com.lansetech.framework.redis.peak.shaving;

import org.springframework.data.redis.connection.RedisConnection;
import redis.clients.util.SafeEncoder;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

public class PeakingShavingPatrol extends Thread {

    private final RedisConnection connection;
    private final PeakShavingRunnableContext context;
    private long lastRefreshTime;
    private List<PeakShavingRunnable> runners;

    public PeakingShavingPatrol(PeakShavingRunnableContext context) {
        super("thread-peaking-shaving-patrol");
        this.context = context;
        this.connection = context.getConnectionFactory().getConnection();

    }

    @Override
    public void run() {
        super.run();
        BlockingQueue<PeakShavingRunnable> queues = context.getAwaitBlockingQueue();
        BlockingQueue<PeakShavingRunnable> emptyQueue = new LinkedBlockingQueue<>();
        while (context.getState() != RunnableState.STOP) {
            try {
                PeakShavingRunnable runner;
                // 先尝试获取等待被唤醒的队列，如果等待队列里面没有
                if ((runner = queues.poll(1500, TimeUnit.MILLISECONDS)) == null) {
                    // 如果等待唤醒的队列是空的，但是还有一些队列之前已经空了，那就检查一下
                    // 看看是不是有队列已经有数据了，有的话就先唤醒他们
                    if (!emptyQueue.isEmpty()) {
                        int size = emptyQueue.size();
                        for (int i = 0; i < size; i++) {
                            PeakShavingRunnable emptyRunner = emptyQueue.take();
                            refreshRunner(emptyRunner);
                            boolean isQueueEmpty = emptyRunner.types().isEmpty();
                            if (isQueueEmpty) {
                                emptyQueue.put(emptyRunner);
                            } else {
                                wakeup(emptyRunner.runnerId());
                            }
                        }
                    }
                    if (needToRefresh()) {
                        refreshQueues();
                    }
                    continue;
                }
                refreshRunner(runner);
                boolean isQueueEmpty = runner.types().isEmpty();
                // queue有数据了，唤醒该队列，让队列继续干活儿
                if (!isQueueEmpty) {
                    wakeup(runner.runnerId());
                } else {
                    emptyQueue.put(runner);
                }
                if (needToRefresh()) {
                    refreshQueues();
                }
            } catch (InterruptedException ignore) {
            }
        }
        // 尝试唤醒所有队列
        for (PeakShavingRunnable runner : runners) {
            wakeup(runner.runnerId());
        }
        connection.close();
        this.context.getShutdownLatch().countDown();
    }

    private void refreshRunner(PeakShavingRunnable runner) {
        // 是否有数据
        String prefix = context.getConfig().getQueuePrefix();
        String pattern = prefix + ":" + PeakShavingConfig.PEAK_SHAVING_TAG_NORMAL + ":*:" + runner.runnerId();
        Set<byte[]> keys = this.connection.keys(SafeEncoder.encode(pattern));
        if (keys == null) {
            keys = new HashSet<>();
        }
        Set<String> newKeys = new HashSet<>();
        for (byte[] key : keys) {
            String keyStr = SafeEncoder.encode(key);
            String[] segs = keyStr.split(":");
            if (segs.length < 4) {
                continue;
            }
            String type = segs[segs.length - 2];
            if (type == null || "".equals(type.trim())) {
                continue;
            }
            newKeys.add(type);
        }
        boolean update = needToUpdate(runner.types(), newKeys);
        if (update) {
            runner.resetTypes(newKeys);
        }
    }

    private void wakeup(int runnerId) {
        Lock lock = context.obtainLock(runnerId);
        Condition condition = context.obtainLockCondition(runnerId);
        lock.lock();
        condition.signalAll();
        lock.unlock();
    }

    public boolean needToRefresh() {
        return System.currentTimeMillis() - lastRefreshTime >= 1000;
    }

    public void registerQueues(List<PeakShavingRunnable> runners) {
        if (runners == null || runners.isEmpty()) {
            this.runners = new LinkedList<>();
        } else {
            this.runners = runners;
        }
    }

    public void refreshQueues() {
        if (this.runners == null) {
            throw new RuntimeException("未注册队列！");
        }
        String prefix = context.getConfig().getQueuePrefix();
        Set<byte[]> keys = this.connection.keys(SafeEncoder.encode(prefix + ":" + PeakShavingConfig.PEAK_SHAVING_TAG_NORMAL + ":*"));
        Map<Integer, List<String>> runnerKeysMap = new HashMap<>();
        for (byte[] keyBytes : keys) {
            String key = SafeEncoder.encode(keyBytes);
            String[] seg = key.split(":");
            // 考虑意外情况
            if (seg.length < 4) {
                continue;
            }
            String indexStr = seg[seg.length - 1];
            int index;
            try {
                index = Integer.parseInt(indexStr);
            } catch (NumberFormatException nfe) {
                continue;
            }
            String type = seg[seg.length - 2];
            if (type == null || type.trim().length() == 0) {
                continue;
            }
            List<String> groupKeys = runnerKeysMap.get(index);
            if (groupKeys == null) {
                groupKeys = new LinkedList<>();
                runnerKeysMap.put(index, groupKeys);
            }
            groupKeys.add(type);
        }

        for (PeakShavingRunnable runner : runners) {
            int rid = runner.runnerId();
            Collection<String> oldTypes = runner.types();
            Collection<String> newTypes = runnerKeysMap.get(rid);
            boolean update = needToUpdate(oldTypes, newTypes);
            if (!update) {
                continue;
            }
            if (!runnerKeysMap.containsKey(rid)) {
                runner.resetTypes(new HashSet<String>());
            } else {
                runner.resetTypes(new HashSet<>(runnerKeysMap.get(rid)));
            }
        }

        lastRefreshTime = System.currentTimeMillis();
    }

    private boolean needToUpdate(Collection<String> oldTypes, Collection<String> newTypes) {
        if (oldTypes == null) {
            oldTypes = new ArrayList<>();
        }
        List<String> type1 = new ArrayList<>(oldTypes);
        if (newTypes == null) {
            newTypes = new ArrayList<>();
        }
        List<String> type2 = new ArrayList<>(newTypes);
        if (type1.size() != type2.size()) {
            return true;
        }
        Collections.sort(type1, new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
                return o1.compareTo(o2);
            }
        });
        Collections.sort(type2, new Comparator<String>() {
            @Override
            public int compare(String o1, String o2) {
                return o1.compareTo(o2);
            }
        });
        for (int i = 0; i < type1.size(); i++) {
            String s1 = type1.get(i);
            String s2 = type2.get(i);
            if (!s1.equals(s2)) {
                return true;
            }
        }
        return false;
    }
}
